/**
 * AMQP接收消息
 *
 * Created by snowingsea on 2020/03/03.
 */
const os = require('os');
const path = require('path');
const amqp = require('amqplib');
const fs = require('fs-extra');
const debug = require('debug')('phoenix:amqp');
const EventEmitter = require('events');
const { ObjectId } = require('mongodb');
const logger = require('../logger');
const projectVersions = require('./../project-versions');
const Func = require('../mongo/function');
const projects = require('./../projects');
const files = require('./../files');
const functions = require('./../functions');
const fileElements = require('./../file-elements');
const tests = require('./../tests');
const testCaseGroups = require('./../test-case-groups');
const tasks = require('./../tasks');
const utility = require('./../utility');
const analyzeTemplates = require('./../analyze-templates');
const rules = require('./../rules');
const defects = require('./../defects');
const invokeRelations = require('../invoke-relations');
const functionRelations = require('../function-relations');
const cacheUtil = require('../../cacheUtil');
const cls = require('../cls');
const TaskLog = require('../mongo/task-log');
const Test = require('../mongo/test-case');
const TestGroup = require('../mongo/test-case-groups');
const File = require('../mongo/file');
const DataFlow = require('../mongo/data-flow');
const { publishSingleTaskMessage, publishValidationMessage } = require('../redis/publisher');
const { kind: fileKind } = require('../files');
const { runTimeCheckRules } = require('../runtime-check');
const RuntimeCheck = require('../mongo/runtime-check');
const codeInstruments = require('../code-instrumentation');
const UserDefinedFields = require('../mongo/user-defined-fields');
const FileTypes = require('../mongo/file-type-system');
const breakpoint = require('../break-points');
const FileTypeSystem = require('../mongo/file-type-system');
const { clusterConfig, workerKind } = require('../../cluster');
const testbeds = require('../testbeds');

const MAX_TRY_CNT = 2; // 消息处理最大尝试次数

const taskCache = new Map();

class MessageBus extends EventEmitter {
  constructor() {
    super();
    this.connection = null;
    this.connPromise = null;
    this.channel = null;
    this.chPromise = null;

    this.queueMap = new Map();
  }

  /**
   * 单例模式
   * @returns {MessageBus}
   */
  static getInstance() {
    if (!MessageBus.instance) {
      MessageBus.instance = new MessageBus();
    }
    return MessageBus.instance;
  }

  static getAMQPURL() {
    const {
      host,
      port,
      username,
      password,
    } = global.config.amqp;
    return `amqp://${username}:${password}@${host}:${port}/rocket`;
  }

  init() {
    if (!this.chPromise) {
      this.chPromise = this.connect()
        .then((conn) => conn.createChannel())
        .then((channel) => {
          this.channel = channel;
          return Promise.resolve();
        })
        .catch((err) => {
          this.reset();
          throw err;
        });
    }
    return this.chPromise;
  }

  connect() {
    if (!this.connPromise) {
      this.connPromise = amqp.connect(MessageBus.getAMQPURL())
        .then((conn) => {
          this.connection = conn;
          conn.on('error', (err) => {
            logger.error('MQ connection error', err);
            this.reset();
            this.emit('connection error');
          });
          return conn;
        })
        .catch((err) => {
          this.reset();
          throw err;
        });
    }
    return this.connPromise;
  }

  sendToQueue(queue, msg, options) {
    return this.init()
      .then(() => this.channel.assertQueue(queue))
      .then(() => this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(msg)), options))
      .catch((err) => {
        logger.error(`Fail to send message to AMQP: ${err}`);
        throw err;
      });
  }

  async createDurableQueue(queue) {
    logger.info('createDurableQueue', queue);
    await this.init();
    await this.channel.assertQueue(queue);
  }

  async createTaskQueue(taskId) {
    // eslint-disable-next-line no-use-before-define
    const queue = getTaskQueue(taskId);
    await this.createDurableQueue(queue);
    await this.channel.bindQueue(queue, 'rocket-exchange', taskId.toString());
  }

  async removeQueue(queue) {
    logger.info('removeQueue', queue);
    await this.init();
    await this.unlisten(queue);
    await this.channel.deleteQueue(queue);
  }

  listen(queue, callback, prefetch = 1) {
    if (this.queueMap.has(queue)) {
      return logger.warn(`queue ${queue} already being consumed`);
    }
    return this.init()
      .then(() => {
        this.channel.prefetch(prefetch);
        return this.channel.assertQueue(queue);
      })
      .then(() => this.channel.consume(
        queue,
        (msg) => callback(msg).finally(() => {
          if (msg) {
            this.channel.ack(msg);
          }
        })
      ))
      .then((consume) => {
        const { consumerTag } = consume;
        this.queueMap.set(queue, {
          consumerTag,
        });
      });
  }

  async listenTaskQueue(queue) {
    logger.info('listenTaskQueue', queue);
    await this.listen(queue, async (msg) => {
      if (msg) {
        cls.createRequestContext({ id: queue });
        try {
          // eslint-disable-next-line no-use-before-define
          await handleMessage(msg);
        } catch (err) {
          const messageType = msg.properties.headers['message-type'];
          const taskId = msg.properties.headers['task-id'];
          logger.error(`Fail to handle message\ntaskId: ${taskId}\nmessageType: ${messageType}\nerror: ${err.stack}`);
        }
      }
    });
  }

  /**
   * 停止消费某个队列
   * @param {string} queue
   */
  async unlisten(queue) {
    if (this.queueMap.has(queue)) {
      logger.info('unlisten', queue);
      const { consumerTag } = this.queueMap.get(queue);
      await this.channel.cancel(consumerTag);
      this.queueMap.delete(queue);
    } else {
      // logger.warn(`queue ${queue} not being consumed`);
    }
  }

  disconnect() {
    return Promise.resolve()
      .then(() => {
        if (!this.channel) {
          return Promise.resolve();
        }
        return this.channel.close();
      })
      .catch(() => { })
      .then(() => {
        if (!this.connection) {
          return Promise.resolve();
        }
        return this.connection.close();
      })
      .catch(() => { })
      .then(() => {
        this.reset();
        return Promise.resolve();
      });
  }

  reset() {
    this.connection = null;
    this.connPromise = null;
    this.channel = null;
    this.chPromise = null;
  }
}
/**
 * 检查连接情况
 * @return {Promise}
 */
function checkConnection() {
  return new Promise((resolve) => {
    MessageBus.getInstance().disconnect()
      .then(() => MessageBus.getInstance().connect())
      .then(() => {
        const { host, port } = global.config.amqp;
        debug(`Successfully Connect to AMQP(${host}:${port})`);
        return MessageBus.getInstance().disconnect();
      })
      .then(resolve)
      .catch((err) => {
        logger.error(`Fail to connect AMQP: ${err}`);
        debug('The connection will be tried again in 10 seconds...');
        setTimeout(() => checkConnection().then(resolve), 10000);
      });
  });
}

/**
 * 处理编译器消息
 * @param {String} taskId 任务id
 * @param {String} projectVersionId 项目版本id
 * @param {Object} content 消息正文内容
 * @return {Promise}
 */
function handleDiagMessage(taskId, projectVersionId, content) {
  return new Promise((resolve, reject) => {
    let file;
    let fileId;
    files.getFileByPath(projectVersionId, content.file)
      .then((res) => {
        file = res;
        if (!file || file.parsingTaskId !== taskId) {
          return Promise.resolve();
        }
        fileId = file._id;
        const { versionId } = file;
        let compileFailed = false;
        content.diag.forEach(({ type }) => {
          switch (type) {
            case 'error':
            case 'fatal error':
              compileFailed = true;
              break;
            default:
              break;
          }
        });
        const status = compileFailed ? files.statusKind.compileFailed : files.statusKind.compiled;
        const promises = [files.updateFileStatus(fileId, status)];
        if (content.includes) {
          promises.push(files.updateFileIncludes(fileId, versionId, content.includes.files));
        }
        return Promise.all(promises);
      })
      .then(() => files.replaceDiag(file.projectId, projectVersionId, fileId, content.diag, content.includes.files, content.file))
      .then(resolve)
      .catch(reject);
  });
}

async function handleProveMessage(taskId, projectVersionId, content, taskInfo) {
  const ruleIds = new Set(taskInfo.ruleIds);
  projectVersionId = ObjectId(projectVersionId);
  const filePaths = new Set();
  content.Result.forEach((item) => {
    item.file = files.getNormalizePath(item.file);
    filePaths.add(item.file);
    item.paths.forEach((arr) => {
      arr.forEach((p) => {
        const [file, line, column] = p.location.split(':');
        p.file = files.getNormalizePath(file);
        p.line = line;
        p.column = column;
        if (p.url) {
          filePaths.add(p.file);
        }
      });
    });
  });
  const fileArr = await File.find(
    {
      versionId: projectVersionId,
      path: {
        $in: [...filePaths],
      },
      kind: fileKind.file,
    }
  );

  const list = content.Result.map((item) => {
    let fileId;
    const fileObj = fileArr.find((f) => f.path === item.file);
    if (fileObj) {
      fileId = fileObj._id;
    }
    const paths = item.paths.map((arr) => arr.map((p) => {
      let fileId;
      const fileObj = fileArr.find((f) => f.path === p.file);
      if (fileObj) {
        fileId = fileObj._id;
      }
      return {
        note: p.note,
        fileId,
        filePath: p.file,
        line: p.line,
        column: p.column,
      };
    }));
    return {
      fileId,
      filePath: item.file,
      line: item.line,
      column: item.column,
      message: item.message,
      paths,
      rule: item.rule,
      versionId: projectVersionId,
    };
  }).filter((item) => {
    item.rule = runTimeCheckRules[item.rule];
    return item.rule && ruleIds.has(item.rule);
  });
  await RuntimeCheck.insertMany(list, { ordered: false }).catch((err) => {
    if (err.writeErrors) {
      const resErrors = err.writeErrors.filter(({ code }) => code !== 11000);
      if (resErrors.length > 0) {
        throw resErrors[0];
      }
    }
  });
}

async function appendUsedTypes(content, fileId) {
  if (!content.functions || !utility.isArrayType(content.functions)) {
    return null;
  }
  const appendTypes = {};
  content.functions.forEach((func) => {
    if (utility.isEmpty(func['@scopes'])) {
      return;
    }
    func['@scopes'].forEach((scope) => {
      if (scope['@kind'].indexOf('RECORD') !== -1) {
        appendTypes[scope['@type']] = 1;
      }
    });
  });
  if (utility.isEmpty(appendTypes)) {
    return null;
  }
  let lastId = null;
  let next = true;
  const usedTypes = {};
  const split = 5;
  const filter = { fileId: ObjectId(fileId) };
  while (next) {
    if (lastId) {
      filter._id = { $gt: lastId };
    }
    // eslint-disable-next-line no-await-in-loop
    const fileTypes = await FileTypes
      .find(filter)
      .sort({ _id: 1 })
      .limit(split)
      .lean();
    fileTypes.forEach((fileType) => {
      Object.keys(appendTypes).forEach((key) => {
        if (fileType.types && fileType.types[key]) {
          usedTypes[key] = fileType.types[key];
          delete appendTypes[key];
        }
      });
    });
    next = fileTypes.length === split && Object.keys(appendTypes).length > 0;
    if (next) {
      lastId = fileTypes[fileTypes.length - 1]._id;
    }
  }
  return usedTypes;
}

/**
 * 处理testif消息
 * @param {String} taskId 任务id
 * @param {String} versionId 项目版本id
 * @param {Object} content 消息正文内容
 * @return {Promise}
 */
async function handleTestIFMessage(taskId, versionId, content) {
  publishSingleTaskMessage('update', {
    taskId: content.phoenixTaskId,
    currentFile: path.basename(content.file),
    subTaskType: tasks.type.parse,
  });
  if (!content.functions) {
    content.functions = [];
  }
  let firstParse;
  let integrationTest = false;
  const updateInvokeRelations = true;
  const updatedElements = [];
  const file = await files.getFileByPath(versionId, content.file);
  if (!file || file.parsingTaskId !== taskId) {
    return;
  }
  const version = await projectVersions.getVersion(versionId);
  if (!version) {
    return;
  }
  // 集成测试解析增加版本锁
  if (version.versionType === projectVersions.versionType.integrationTest) {
    integrationTest = true;
  }
  // 回归版本要针对老版本的相应子版本进行回归
  let { oldVersionId } = version;
  if (oldVersionId && version.versionType !== projectVersions.versionType.unitTest) {
    const relationVersions = await projectVersions.getRelationVersions(oldVersionId);
    relationVersions.forEach((relationVersion) => {
      if (relationVersion.versionType === version.versionType) {
        oldVersionId = relationVersion._id;
      }
    });
  }
  let contentGroupByFile;
  if (!integrationTest) {
    // 单元测试仅过滤出本文件的函数
    contentGroupByFile = [{
      fileId: file._id.toString(),
      insideProject: true,
      contentFuncs: content.functions.filter((func) => func.location.path === file.path),
    }];
  } else {
    // 根据文件对函数进行分组
    contentGroupByFile = await files.getContentGroupByFile(content.functions, file);
    await fileElements.removeSourceParsedFile(file._id, content.functions, versionId);
  }
  delete content.functions;
  const templateInfo = new Map();
  await utility.arrayChunkOperation(contentGroupByFile, 1, async ({ contentFuncs, fileId, insideProject }) => {
    content.functions = contentFuncs;
    // 拼接解析所需要的类型
    content.types = await appendUsedTypes(content, fileId);
    const newElements = fileElements.analyzeFunctionContent(content);
    let funcs = await functions.getFunctions(versionId, {}, fileId);

    firstParse = funcs.length === 0 && file.status !== files.statusKind.waiting4compiling;
    // 对于其他文件来说此字段无意义，设置为true使后续的unread判断无效
    if (fileId !== file._id.toString()) {
      firstParse = true;
    }
    // 本文件函数，比较删除需要删除的函数
    if (fileId === file._id.toString()) {
      await fileElements.removeOldFileElements(newElements, funcs, versionId);
    }
    // 对于项目内文件更新函数hash
    if (insideProject) {
      // 读取文件注释信息
      await files.readFileComment(fileId.toString(), versionId);
      // compute function hash and change kind
      funcs = [...newElements.values()].filter(
        (fun) => fun.kind === fileElements.elementKind.function || fun.kind === fileElements.elementKind.templateFunction
          || fun.kind === fileElements.elementKind.specialFunction
      );
      const ranges = funcs.map((fun) => [fun.locationRange.begin.line * 1, fun.locationRange.end.line * 1]);
      const lines = await files.getLinesOfContentBatch(file._id, ranges);
      lines.forEach((lineArr, idx) => {
        const func = funcs[idx];
        if (!lineArr || !lineArr.length) {
          // 如果内容为空，不计算hash
          func.hash = undefined;
        } else {
          // note: 核心层给的column是光标位置，不是字符位置
          // 先截取末尾，再截取开头，顺序不能乱
          lineArr[lineArr.length - 1] = lineArr[lineArr.length - 1].slice(0, func.locationRange.end.column * 1);
          lineArr[0] = lineArr[0].slice(func.locationRange.begin.column - 1);
          func.hash = utility.md5(lineArr.join('\n'));
        }
      });
      if (oldVersionId) {
        // 回归版本判断函数变更状态
        newElements.forEach((func) => {
          func.changed = 'added';
        });
        const oldFile = await files.getFileByPath(oldVersionId, file.path);
        if (oldFile) {
          const oldFuncs = await Func.find({ fileId: oldFile._id });
          oldFuncs.forEach((func) => {
            const newFunc = newElements.get(func.mangledId);
            if (newFunc) {
              if (newFunc.hash !== func.hash) {
                newFunc.changed = 'modified';
              } else {
                newFunc.changed = undefined;
              }
            }
          });
        }
      }
      if (content.lineStatistic) {
        await File.updateOne(
          { _id: file._id },
          {
            $set: {
              numberOfLines: content.lineStatistic.total || 0,
              sourceLines: content.lineStatistic.code || 0,
              commentLines: content.lineStatistic.comment || 0,
              emptyLines: content.lineStatistic.blank || 0,
            },
          }
        );
      }
    }
    // for now we will merge new element data with exists specialized data together
    // to update the 'index' and send ws message
    // so retrieve original data again is because above function may delete
    // some related specialized data
    funcs = await functions.getFunctions(versionId, {}, fileId);
    // append extra info to new elements like elements exist flag and specialized info
    await fileElements.prepareNewElements(newElements, funcs);
    await fileElements.replaceFileElements(
      newElements,
      {
        fileId,
        projectId: file.projectId,
        versionId,
        creatorId: file.creatorId,
        firstParse,
        sourceParsedFile: file._id,
        insideProject,
      }
    );
    // 准备特化相关数据
    newElements.forEach((value, key) => {
      if (!templateInfo.has(key)) {
        templateInfo.set(key, { id: value._id, parent: value.parent });
      }
      if (value.template || value.rawTemplate) {
        templateInfo.get(key).template = value.template || '';
        templateInfo.get(key).rawTemplate = value.rawTemplate || '';
      }
    });
    // if new created element under a template element, need to delete the
    // specialized data which related with that template element
    await fileElements.removeSpecializedDataForNewElements(newElements);
    // 集成测试更新函数调用关系
    if (updateInvokeRelations) {
      let queue = Promise.resolve();
      let relationUpdate;
      content.functions.forEach((item) => {
        queue = queue.then(
          () => {
            relationUpdate = false;
            const newElement = newElements.get(item['@mangled']);
            if (!fileElements.isTestableElement(newElement.kind)) {
              return Promise.resolve();
            }
            if (newElement.exists) {
              const deleteStubs = new Set();
              const createStubs = new Set();
              return functions.getFunctionVariables(newElement._id)
                .then((functionVariable) => {
                  if (!functionVariable) {
                    return Promise.resolve();
                  }
                  const oldStubs = functionVariable.stubs || {};
                  const newStubs = item.stubs || {};
                  Object.keys(newStubs).forEach((key) => {
                    if (!oldStubs[key]) {
                      createStubs.add({
                        mangledId: newStubs[key]['@mangled'],
                        name: newStubs[key]['@name'],
                        toFuncPointer: !newStubs[key]['@isCall'],
                      });
                      relationUpdate = true;
                    }
                  });
                  Object.keys(oldStubs).forEach((key) => {
                    if (!newStubs[key]
                      && oldStubs[key].stubsTriggerKind === tests.stubTriggerKind.parse) {
                      deleteStubs.add(oldStubs[key]['@mangled']);
                    }
                  });
                  if (functions.diffVariables(
                    JSON.parse(utility.toFullWidth(JSON.stringify(functionVariable.toJSON()))),
                    JSON.parse(utility.toFullWidth(JSON.stringify(item)))
                  )) {
                    updatedElements.push(newElement._id);
                  }
                  return Promise.all([...createStubs].map(
                    (stub) => functionRelations.createRelations(
                      newElement._id,
                      item['@mangled'],
                      fileId,
                      stub.mangledId,
                      stub.name,
                      stub.toFuncPointer,
                      versionId
                    )
                  ));
                })
                .then(() => Promise.all(
                  [...deleteStubs].map((deleteStub) => invokeRelations.removeRelation(
                    newElement._id,
                    deleteStub,
                    versionId,
                    false
                  ))
                ));
            }
            updatedElements.push(newElement._id);
            relationUpdate = true;
            if (utility.isEmpty(item.stubs)) {
              return functionRelations.createRelations(
                newElement._id,
                item['@mangled'],
                fileId,
                '',
                '',
                null,
                versionId,
              );
            }
            return Promise.all(
              Object.values(item.stubs).map((stub) => functionRelations.createRelations(
                newElement._id,
                item['@mangled'],
                fileId,
                stub['@mangled'],
                stub['@name'],
                !stub['@isCall'],
                versionId
              ))
            );
          }
        )
          .then(() => {
            const newElement = newElements.get(item['@mangled']);
            if (!relationUpdate) {
              return Promise.resolve();
            }
            return invokeRelations.createRelations(
              versionId,
              newElement._id,
              item['@mangled'],
              fileId
            );
          });
      });
      await queue;
    }
    // 添加函数变量
    utility.arrayChunkOperation(content.functions, 10, async (item) => {
      await functions.replaceFunctionVariables(
        file.projectId,
        versionId,
        fileId,
        newElements.get(item['@mangled'])._id,
        item,
        // integrationTest
      );
    });
    // 更新函数注释信息
    const newFuncs = await Func.find({ fileId, kind: { $in: fileElements.getCountableKinds() } }, ['_id', 'userDefinedFields', 'location', 'fileId', 'index']).lean();
    let fileFuncsLocation = [];
    newFuncs.forEach((item) => {
      fileFuncsLocation.push(item.location.line);
    });
    fileFuncsLocation = fileFuncsLocation.sort();
    const userDefinedField = await UserDefinedFields.findOne({ domain: version.versionType, scope: 'function', name: '函数注释' });
    await utility.arrayChunkOperation(newFuncs, 50, (func) => functions.updateFunctionComment(
      func,
      fileFuncsLocation,
      userDefinedField
    ));
  });
  // 添加全局变量、字面量、宏替换信息
  await Promise.all([
    files.replaceGlobalVariables(file.projectId, versionId, file._id, content.globals),
    files.replaceLiterals(file.projectId, versionId, file._id, content.literals),
    files.replaceMacro(file.projectId, versionId, file._id, content.macros ? content.macros.usages : []),
  ]);
  // 特化数据更新rawTemplateId,specialOriginalId
  await fileElements.batchUpdateTemplateInfo(templateInfo);
  if (!taskCache.has(taskId.toString())) {
    taskCache.set(taskId.toString(), {});
  }
  taskCache.get(taskId.toString()).updatedElements = updatedElements;
}

async function handleCfgMessage(taskId, version, content) {
  if (!content.functions) {
    content.functions = [];
  }
  const file = await files.getFileByPath(version._id, content.file);
  if (!file || file.parsingTaskId !== taskId) {
    return;
  }
  let contentGroupByFile;
  if (version.versionType !== projectVersions.versionType.integrationTest) {
    // 单元测试仅过滤出本文件的函数
    contentGroupByFile = [{
      fileId: file._id.toString(),
      insideProject: true,
      contentFuncs: content.functions.filter((func) => func.location.path === file.path),
    }];
  } else {
    // 根据文件对函数进行分组
    const fileMap = new Map();
    const filesInDbMap = new Map();
    const filesInDb = await File.find(
      { versionId: version._id, kind: { $in: [fileKind.file, fileKind.fileOutsideProject] } }, ['path', 'kind']
    );
    filesInDb.forEach((file) => {
      filesInDbMap.set(file.path, {
        fileId: file._id.toString(),
        insideProject: file.kind !== fileKind.fileOutsideProject,
      });
    });
    content.functions.forEach((func) => {
      if (!filesInDbMap.has(func.location.path)) {
        return;
      }
      const fileInDb = filesInDbMap.get(func.location.path);
      if (!fileMap.has(func.location.path)) {
        fileMap.set(func.location.path, {
          contentFuncs: [],
          fileId: fileInDb.fileId,
          insideProject: fileInDb.insideProject,
        });
      }
      fileMap.get(func.location.path).contentFuncs.push(func);
    });
    contentGroupByFile = [...fileMap.values()];
  }
  delete content.functions;
  await utility.arrayChunkOperation(contentGroupByFile, 1, async ({ contentFuncs, fileId, insideProject }) => {
    content.functions = contentFuncs;
    if (insideProject) {
      // 获取死循环
      const ranges = [];
      const infiniteLoops = [];
      const infiniteLoopEntry = new Map();
      content.functions.forEach((item) => {
        if (!item.blocks) {
          return;
        }
        item.blocks.forEach((block) => {
          if (!block.infiniteloop
            && [
              'DOWHILE',
              'WHILE',
              'FOR_INC',
              'RANGE',
              'EMPTY_INC',
            ].indexOf(block.kind) < 0) {
            return;
          }
          infiniteLoops.push({
            key: `${item.mangled}|${block.id}`,
            begin: block.begin,
            end: block.end,
            kind: block.kind,
          });
          ranges.push([block.begin.line * 1, block.end.line * 1]);
        });
      });
      if (ranges.length > 0) {
        const lines = await files.getLinesOfContentBatch(file._id, ranges);
        lines.forEach((lineArr, idx) => {
          const infiniteLoop = infiniteLoops[idx];
          lineArr[lineArr.length - 1] = lineArr[lineArr.length - 1].slice(0, infiniteLoop.end.column * 1);
          lineArr[0] = lineArr[0].slice(infiniteLoop.begin.column - 1);
          infiniteLoopEntry.set(infiniteLoop.key, `${infiniteLoop.kind}:${lineArr.join(os.EOL)}`);
        });
        // 将死循环label拼回cfg
        content.functions.forEach((item) => {
          if (!item.blocks) {
            return;
          }
          item.blocks.forEach((block) => {
            if (!block.infiniteloop
              && [
                'DOWHILE',
                'WHILE',
                'FOR_INC',
                'RANGE',
                'EMPTY_INC',
              ].indexOf(block.kind) < 0) {
              return;
            }
            if (infiniteLoopEntry.has(`${item.mangled}|${block.id}`)) {
              block.infiniteLoopEntry = infiniteLoopEntry.get(`${item.mangled}|${block.id}`);
            }
          });
        });
      }
    }
    await utility.arrayChunkOperation(content.functions, 10, async (item) => {
      const func = await functions.getFunctionByKey(fileId, item.mangled);
      if (!func) {
        return;
      }
      if (fileElements.isTestableElement(func.kind)
      ) {
        await functions.replaceFunctionCFG(
          file.projectId,
          version._id,
          fileId,
          func._id,
          item,
          version.mcdcKind
        );
      }
    });
  });
  // 集成测试更新覆盖率分母需要在cfg保存以后
  if (version.versionType === projectVersions.versionType.integrationTest
    && taskCache.has(taskId)
    && taskCache.get(taskId).updatedElements
    && taskCache.get(taskId).updatedElements.length > 0) {
    const updateElements = taskCache.get(taskId).updatedElements;
    const funcs = await invokeRelations.getInvokeFunctions(updateElements, version._id);
    await functions.updateInvokeCoverageTotal([...funcs, ...updateElements], version._id);
    delete taskCache.get(taskId).updatedElements;
  }
}

/**
 * 处理文件类型系统消息
 * @param {String} taskId 任务id
 * @param {String} versionId 项目版本id
 * @param {Object} content 消息正文内容
 * @return {Promise}
 */
async function handleFileTypesystem(taskId, versionId, content) {
  if (!content.file) {
    return;
  }
  const file = await files.getFileByPath(versionId, content.file);
  if (content.id === 1) {
    await files.deleteFileOriginalTypeSystem(file.projectId, file.versionId, file._id);
  }
  const doc = {
    projectId: file.projectId,
    versionId: file.versionId,
    fileId: file._id,
    types: content.types,
  };
  const fileTypeSystem = new FileTypeSystem(doc);
  await fileTypeSystem.save();
}
/**
 * 处理静态分析消息
 * @param {String} taskId 任务id
 * @param {String} versionId 项目版本id
 * @param {Object} content 消息正文内容
 * @param {String} ruleSetId ruleSetId
 * @param {Boolean} isHcscliAnalyse 是否国创静态分析
 * @param {Boolean} isCertifyAnalyse 是Certify模块静态分析
 * @return {Promise}
 */
async function handleAnalyzeMessage(taskId, versionId, content, ruleSetId = undefined, isHcscliAnalyse = false, isCertifyAnalyse = false) {
  publishSingleTaskMessage('update', {
    taskId: content.phoenixTaskId,
    currentFile: path.basename(content.file),
    subTaskType: tasks.type.analyze,
  });
  let file = await files.getFileByPath(versionId, content.file);
  if (!file || file.analyzeTaskId !== taskId) {
    return;
  }
  const version = await projectVersions.getVersion(versionId);
  if (!version || !version.analyzeTemplate) {
    return;
  }
  const template = await analyzeTemplates.getTemplate(version.analyzeTemplate.toString());
  if (!template) {
    return;
  }
  const rulesResult = await rules.getRulesByIds(template.rules, ruleSetId);
  const rulesMap = new Map();
  rulesResult.forEach((rule) => {
    rule.ruleId = rule._id;
    // hcscli部分规则集使用hcscliExternalRuleCode标识规则来源
    if (isHcscliAnalyse && rule.hcscliExternalRuleCode) {
      rule.externalRuleCode = rule.hcscliExternalRuleCode;
    }
    // certify模块解析
    if (isCertifyAnalyse) {
      rule.externalRuleCode = rule.ruleCode;
    }
    if (!rulesMap.has(rule.externalRuleCode)) {
      rulesMap.set(rule.externalRuleCode, []);
    }
    rules.formatRulesForUserDefined(rule, template.userDefined);
    rulesMap.get(rule.externalRuleCode).push(rule);
  });
  const detects = [];
  let originalVersionDefects = [];
  const originalDefectsMap = new Map();
  // 处理回归版本一键回归信息
  const taskType = (await tasks.getTaskByHydraTaskId(taskId)).type;
  if (version.oldVersionId && taskType === tasks.type.regressionAnalyze) {
    // 获取原始静态分析版本id
    const { _id: staticAnalyzeVersionId } = await projectVersions.initSubVersion(
      version.oldVersionId,
      projectVersions.versionType.staticAnalyze
    );
    originalVersionDefects = await defects.getDefects(staticAnalyzeVersionId,
      { filePath: content.file, includeOriginal: true });
    originalVersionDefects.forEach((defect) => {
      const localtions = defect.locations;
      let startLine = 0;
      let startColumn = 0;
      if (localtions.begin) {
        startLine = localtions.begin.line ? localtions.begin.line : 0;
        startColumn = localtions.begin.column ? localtions.begin.column : 0;
      }
      originalDefectsMap.set(`${defect.ruleId.toString()}-${startLine}-${startColumn}`, defect);
    });
  }
  const ruleSetIds = new Set();
  content.Result.forEach((defect) => {
    if (!rulesMap.has(defect.bug_type)) {
      return;
    }
    rulesMap.get(defect.bug_type).forEach((rule) => {
      const defectObj = {
        projectId: file.projectId,
        versionId: file.versionId,
        fileId: file._id,
        filePath: file.path,
        ruleId: rule._id,
        ruleSetId: rule.ruleSetId,
        status: defects.defectStatus.toBeEdit,
        locations: defect.locations,
        level: rule.level,
        originalLevel: rule.originalLevel,
        subLevelOrder: rule.subLevelOrder,
        message: defect.message,
        defectType: defects.defectType.add,
        notes: defect.notes,
      };
      ruleSetIds.add(rule.ruleSetId.toString());
      if (defect.steps) {
        defectObj.steps = defect.steps.map((step) => ({
          filePath: files.getNormalizePath(step.file_path),
          line: step.line,
        }));
      }
      if (taskType === tasks.type.regressionAnalyze && originalVersionDefects.length > 0) {
        const localtions = defect.locations;
        let startLine = 0;
        let startColumn = 0;
        if (localtions.begin) {
          startLine = localtions.begin.line ? localtions.begin.line : 0;
          startColumn = localtions.begin.column ? localtions.begin.column : 0;
        }
        if (originalDefectsMap.has(`${defectObj.ruleId.toString()}-${startLine}-${startColumn}`)) {
          defectObj.defectType = defects.defectType.original;
        }
        detects.push(
          defectObj
        );
      } else {
        detects.push(defectObj);
      }
    });
  });
  file = await files.getFileById(file._id);
  await defects.bulkInsert(detects, file);
}

async function handleTargetMachineExport(taskInfo, projectId, versionId, testId, ackNo, content) {
  let testcase = await tests.getTest(testId);
  let func;
  if (!testcase) {
    testcase = await testCaseGroups.getGroupById(testId);
    if (testcase.length === 0) {
      return;
    }
    [testcase] = testcase;
    func = await functions.getFunction(testcase.functionId);
    testcase.filepath = func.filepath;
    testcase.versionType = func.versionType;
    testcase.identifierNumber = testcase.number;
  }
  if (!func) {
    func = await functions.getFunction(testcase.functionId);
  }
  testcase.mangledId = func.mangledId;
  const {
    filepath, versionType, identifierNumber,
  } = testcase;
  let { mangledId } = testcase;
  const { codeInstrumentationId, coverageStatistics } = taskInfo;
  const coverageInfo = coverageStatistics ? 'IncludeCoverage' : 'ExcludeCoverage';
  const { projectName } = await projects.getProject(projectId);
  if (process.platform === 'win32') {
    mangledId = mangledId.replace(/[/\\:*?"<>|]/g, '');
  }
  const taskPath = path.join(cacheUtil.getCodeInstrumentationDir(), versionId.toString(), codeInstrumentationId.toString(), `TestScript_${coverageInfo}`);
  const fnPath = `${filepath.replace('./', '').replace(/\.|\//g, '_')}__${mangledId}_${identifierNumber}`;
  const resourcePath = `${__dirname}/../../../resources/target-machine/`;
  if (versionType === 'integrationTest') {
    const p = `${taskPath}/integration-common`;
    if (!await fs.pathExists(p)) {
      await fs.copy(`${resourcePath}/integration-common`, p);
    }
  } else {
    const p = `${taskPath}/unit-common`;
    if (!await fs.pathExists(p)) {
      await fs.copy(`${resourcePath}/unit-common`, p);
    }
  }
  const finalPath = path.join(taskPath, projectName, fnPath);
  await fs.ensureDir(finalPath);
  const targetPath = path.join(finalPath, `testcase_${ackNo}.c`);
  await fs.writeFile(targetPath, content);
}
/**
 *处理规则集映射的消息
 */
function handleAnalyzeByRuleSetMapping(data) {
  const newContentArray = [];
  data.Result = data.Result.filter((r) => [`./${r.file}`, `./${r.File}`].includes(data.file)
    && r.spec && ![r.Check, r.check].includes(r.spec));
  const { Mapping, Result, file } = data;
  if (Result.length === 0) {
    return newContentArray;
  }
  if (!Mapping) {
    const ruleSetName = data.Result[0].spec;
    return [{
      ...data,
      ruleSetName,
      mappingName: `${ruleSetName}:${ruleSetName}`,
    }];
  }
  // 将Result中不存在映射的规则集放到mapping中
  Result.forEach((r) => {
    if (r.spec && !Mapping[r.spec]) {
      Mapping[r.spec] = [r.spec];
    }
  });
  // 分配mapping中的规则集缺陷数据
  Object.keys(Mapping).forEach((mappingName) => {
    Mapping[mappingName].forEach((ruleSetName) => {
      const newResult = [];
      Result.forEach((item) => {
        if (item.spec === mappingName) {
          newResult.push({
            ...item,
            spec: ruleSetName,
          });
        }
      });
      newContentArray.push({
        file,
        Result: newResult,
        mappingName: `${mappingName}:${ruleSetName}`,
        ruleSetName,
      });
    });
  });
  return newContentArray;
}
/**
 * 处理MISRAC-2012，GJB5369-2005规则集静态分析消息
 * @param {String} taskId 任务id
 * @param {String} versionId 项目版本id
 * @param {Object} content 消息正文内容
 *
 * TJ_DW230-2020-C映射到MISRAC-2004
 *
 */
async function handleHcscliAnalyzeMessage(taskId, versionId, data) {
  const { Result, phoenixTaskId } = data;
  if (!Result || Result.length === 0) {
    return;
  }
  const allRules = await rules.getRules();
  if (allRules.length === 0) {
    return;
  }
  // console.log('hcscliResult', JSON.stringify(data));
  const contents = handleAnalyzeByRuleSetMapping(data);
  const promises = [];
  contents.forEach((content) => {
    promises.push(() => {
      const newResult = [];
      const { ruleSetName } = content;
      const ruleSet = allRules.find((r) => r.ruleSetName === ruleSetName);
      if (!ruleSet) {
        return Promise.resolve();
      }
      // 消息结构转换
      content.Result.forEach((item) => {
        // check if the item is valid or not, 2004 and TJ use hcscliExternalRuleCode field
        const rule = ruleSet.rules.find((r) => r.hcscliExternalRuleCode === item.Check
          || (!r.hcscliExternalRuleCode && r.externalRuleCode === item.Check));
        if (!rule) {
          return;
        }
        const locations = {
          begin: {
            line: parseInt(item.Line || item.StartLine, 10),
            column: parseInt(item.StartColumn, 10),
          },
          end: {
            line: parseInt(item.Line || item.EndLine, 10),
            column: parseInt(item.EndColumn, 10),
          },
        };
        const ruleInfo = {
          spec: item.spec,
          bug_id: item.ID,
          bug_type: item.Check,
          file_path: item.File,
          locations,
          message: item.Text,
          steps: [],
        };
        if (item.Notes) {
          item.Notes.forEach((note) => {
            ruleInfo.steps.push({
              column: {
                start: parseInt(note.StartColumn, 10),
                end: parseInt(note.EndColumn, 10),
              },
              file_path: note.File,
              line: parseInt(note.Line || note.StartLine, 10),
              message: note.Text,
            });
          });
        }
        newResult.push(ruleInfo);
      });
      if (newResult.length === 0) {
        return Promise.resolve();
      }
      return handleAnalyzeMessage(taskId, versionId, { file: content.file, Result: newResult, phoenixTaskId }, ruleSet.ruleSetId, true);
    });
  });
  await utility.promiseChain(promises);
}
async function handleScannerAnalyzeMessage(taskId, versionId, data) {
  const { Result, phoenixTaskId } = data;
  if (!Result || Result.length === 0) {
    return Promise.resolve();
  }
  const filePathMap = new Map();
  const filePaths = await File.find({ versionId, kind: { $ne: files.kind.directory } }, ['_id', 'path']);
  filePaths.forEach((file) => {
    filePathMap.set(file.path, file.id.toString());
  });
  const newResult = [];
  // 消息结构转换
  Result.forEach((defect, index) => {
    if (!defect.rule || !defect.spec || !defect.locations || defect.locations.length === 0) {
      return;
    }
    const notes = [];
    const locations = { begin: {}, end: {} };
    defect.locations.forEach(({ physicalLocation }) => {
      if (!physicalLocation) {
        return;
      }
      const { region, artifactLocation } = physicalLocation;
      if (!region || !artifactLocation) {
        return;
      }
      const filePath = `./${artifactLocation.uri}`;
      const {
        startLine, startColumn, endLine, endColumn,
      } = region;
      if (index === 0) {
        locations.begin.line = startLine;
        locations.begin.column = startColumn;
        locations.end.line = endLine;
        locations.end.column = endColumn;
        return;
      }
      if (!filePathMap.has(filePath)) {
        return;
      }
      notes.push({
        fileId: filePathMap.get(filePath),
        filePath,
        fileName: path.basename(filePath),
        locations: {
          begin: {
            line: startLine,
            column: startColumn,
          },
          end: {
            line: endLine,
            column: endColumn,
          },
        },
        description: '',
      });
    });
    if (defect.relatedLocations) {
      defect.relatedLocations.forEach(({ physicalLocation, message }) => {
        if (!physicalLocation || !message) {
          return;
        }
        const { region, artifactLocation } = physicalLocation;
        if (!region || !artifactLocation) {
          return;
        }
        const filePath = `./${artifactLocation.uri}`;
        if (!filePathMap.has(filePath)) {
          return;
        }
        const {
          startLine, startColumn, endLine, endColumn,
        } = region;
        notes.push({
          fileId: filePathMap.get(filePath),
          filePath,
          fileName: path.basename(filePath),
          locations: {
            begin: {
              line: startLine,
              column: startColumn,
            },
            end: {
              line: endLine,
              column: endColumn,
            },
          },
          description: message.text,
        });
      });
    }
    const ruleInfo = {
      spec: defect.spec,
      bug_type: defect.rule,
      file_path: defect.File,
      locations,
      message: defect.message.text,
      notes,
    };
    newResult.push(ruleInfo);
  });
  if (newResult.length === 0) {
    return Promise.resolve();
  }
  return handleAnalyzeMessage(taskId, versionId, { file: data.file, Result: newResult, phoenixTaskId }, null, false);
}
/**
 * 处理Certify静态分析消息
 * @param {String} taskId 任务id
 * @param {String} versionId 项目版本id
 * @param {Object} content 消息正文内容
 */
async function handleCertifyAnalyzeMessage(taskId, versionId, content) {
  const { phoenixTaskId } = content;
  const newResult = [];
  if (!content.Result) {
    return;
  }
  content.Result = content.Result.filter((r) => r.check && `./${r.file}` === content.file && r.spec);
  if (content.Result.length === 0) {
    return;
  }
  const ruleSetName = content.Result[0].spec;
  const allRules = await rules.getRules();
  if (allRules.length === 0) {
    return;
  }
  const ruleSet = allRules.find((r) => r.ruleSetName === ruleSetName);
  if (!ruleSet) {
    return;
  }
  // 消息结构转换
  content.Result.forEach((item) => {
    if (!item.rule) {
      return;
    }
    let ruleCode;
    ruleSet.rules.forEach((rule) => {
      if (rule.ruleCode === item.rule) {
        ruleCode = rule.ruleCode;
      }
    });
    if (!ruleCode) {
      return;
    }
    let fixIt = [];
    let { message } = item;
    let locations = {
      begin: {
        line: null,
        column: null,
      },
      end: {
        line: null,
        column: null,
      },
    };
    if (item['fix-it']) {
      fixIt = item['fix-it'];
      const { range } = fixIt[0];
      if (range) {
        locations = range;
      }
    } else if (item.column && item.line) {
      locations.begin.line = locations.end.line = item.line;
      locations.begin.column = locations.end.column = item.column;
    }
    message = `<${locations.begin.line}:${locations.begin.column};${locations.end.line}:${locations.end.column}>${message}`;
    const ruleInfo = {
      spec: item.spec,
      bug_type: ruleCode,
      file_path: item.file,
      locations,
      message,
    };
    newResult.push(ruleInfo);
  });
  if (newResult.length === 0) {
    return;
  }
  await handleAnalyzeMessage(taskId, versionId, { file: content.file, Result: newResult, phoenixTaskId }, ruleSet.ruleSetId, false, true);
}
/**
 * 处理Verify静态分析消息
 * @param {String} taskId 任务id
 * @param {String} versionId 项目版本id
 * @param {Object} content 消息正文内容
 */
async function handleVerifyAnalyzeMessage(taskId, versionId, content) {
  const { phoenixTaskId } = content;
  const newResult = [];
  if (!content.Result) {
    return;
  }
  content.Result = content.Result.filter((r) => r.check && r.file === content.file && r.spec);
  if (content.Result.length === 0) {
    return;
  }
  const ruleSetName = content.Result[0].spec;
  const allRules = await rules.getRules();
  if (allRules.length === 0) {
    return;
  }
  const ruleSet = allRules.find((r) => r.ruleSetName === ruleSetName);
  if (!ruleSet) {
    return;
  }
  // 消息结构转换
  content.Result.forEach((item) => {
    let ruleCode;
    ruleSet.rules.forEach((rule) => {
      if (rule.ruleCode === item.rule) {
        ruleCode = rule.ruleCode;
      }
    });
    if (!ruleCode) {
      return;
    }
    if (!item.rule) {
      return;
    }
    let fixIt = [];
    let {
      message,
    } = item;
    let locations = {
      begin: {
        line: null,
        column: null,
      },
      end: {
        line: null,
        column: null,
      },
    };
    if (item['fix-it']) {
      fixIt = item['fix-it'];
      const { range } = fixIt[0];
      if (range) {
        locations = range;
      }
    } else if (item.column && item.line) {
      locations.begin.line = locations.end.line = item.line;
      locations.begin.column = locations.end.column = item.column;
    }
    message = `<${locations.begin.line}:${locations.begin.column};${locations.end.line}:${locations.end.column}>${message}`;
    const ruleInfo = {
      spec: item.spec,
      bug_type: ruleCode,
      file_path: content.file,
      locations,
      message,
    };
    newResult.push(ruleInfo);
  });
  if (newResult.length === 0) {
    return;
  }
  await handleAnalyzeMessage(taskId, versionId, { file: content.file, Result: newResult, phoenixTaskId }, ruleSet.ruleSetId, false, true);
}
/**
 * 处理javacheck静态分析消息
 * @param {String} taskId 任务id
 * @param {String} versionId 项目版本id
 * @param {Object} content 消息正文内容
 */
async function handleJavacheckAnalyzeMessage(taskId, versionId, content) {
  const { phoenixTaskId } = content;
  const newResult = [];
  if (!content.Result) {
    return;
  }
  content.Result = content.Result.filter((r) => r.check && r.file === content.file && r.spec);
  if (content.Result.length === 0) {
    return;
  }
  const ruleSetName = content.Result[0].spec;
  const allRules = await rules.getRules();
  if (allRules.length === 0) {
    return;
  }
  const ruleSet = allRules.find((r) => r.ruleSetName === ruleSetName);
  if (!ruleSet) {
    return;
  }
  // 消息结构转换
  content.Result.forEach((item) => {
    let ruleCode;
    ruleSet.rules.forEach((rule) => {
      if (rule.ruleCode === item.rule) {
        ruleCode = rule.ruleCode;
      }
    });
    if (!ruleCode) {
      return;
    }
    if (!item.rule) {
      return;
    }
    let fixIt = [];
    let {
      message,
    } = item;
    let locations = {
      begin: {
        line: null,
        column: null,
      },
      end: {
        line: null,
        column: null,
      },
    };
    if (item['fix-it']) {
      fixIt = item['fix-it'];
      const { range } = fixIt[0];
      if (range) {
        locations = range;
      }
    } else if (item.column && item.line) {
      locations.begin.line = locations.end.line = item.line;
      locations.begin.column = locations.end.column = item.column;
    }
    message = `<${locations.begin.line}:${locations.begin.column};${locations.end.line}:${locations.end.column}>${message}`;
    const ruleInfo = {
      spec: item.spec,
      bug_type: ruleCode,
      file_path: content.file,
      locations,
      message,
    };
    newResult.push(ruleInfo);
  });
  if (newResult.length === 0) {
    return;
  }
  await handleAnalyzeMessage(taskId, versionId, { file: content.file, Result: newResult, phoenixTaskId }, ruleSet.ruleSetId, false, true);
}
/**
 * 处理cppcheck静态分析消息
 * @param {String} taskId 任务id
 * @param {String} versionId 版本id
 * @param {Object} content 消息内容
 *
 * MISRAC-2004映射到MISRAC-2012
 * TJ_DW230-2020-C映射到MISRAC-2012
 *
 */
async function handleCppCheckAnalyzeMessage(taskId, versionId, data) {
  const { Result, phoenixTaskId } = data;
  if (!Result || Result.length === 0) {
    return;
  }
  const allRules = await rules.getRules();
  if (allRules.length === 0) {
    return;
  }
  const contents = handleAnalyzeByRuleSetMapping(data);
  const promises = [];
  contents.forEach((content) => {
    promises.push(() => {
      const newResult = [];
      const { ruleSetName } = content;
      const ruleSet = allRules.find((r) => r.ruleSetName === ruleSetName);
      if (!ruleSet) {
        return Promise.resolve();
      }
      // 消息结构转换
      content.Result.forEach((item) => {
        const rule = ruleSet.rules.find((r) => r.externalRuleCode === item.check);
        if (!rule) {
          return;
        }
        const locations = {
          begin: {
            line: item.line,
            column: item.column,
          },
          end: {
            line: item.line,
            column: null,
          },
        };
        const ruleInfo = {
          spec: item.spec,
          bug_type: item.check,
          file_path: item.file,
          locations,
          message: item.message,
        };
        newResult.push(ruleInfo);
      });
      if (newResult.length === 0) {
        return Promise.resolve();
      }
      return handleAnalyzeMessage(taskId, versionId, { file: content.file, Result: newResult, phoenixTaskId }, ruleSet.ruleSetId);
    });
  });
  await utility.promiseChain(promises);
}
/**
 * 处理测试用例消息
 * @param {String} version 项目
 * @param {Object} content 消息正文内容
 * @return {Promise}
 */
function handleTestCaseMessage(version, content) {
  publishSingleTaskMessage('update', {
    taskId: content.phoenixTaskId,
    currentFile: path.basename(content.file),
    subTaskType: tasks.type.testCaseGenerate,
  });
  const isIntegrationTest = version.versionType && version.versionType === projectVersions.versionType.integrationTest;
  return new Promise((resolve, reject) => {
    let fileId;
    const functionMap = new Map();
    const functionVariablesMap = new Map();
    files.getFileByPath(version._id, content.file)
      .then((file) => {
        if (!file) {
          content.functions = [];
        } else {
          fileId = file._id;
        }
        const promises = content.functions.map((item) => {
          const { name } = item;
          return functions.getFunctionByName(fileId, name);
        });
        return Promise.all(promises);
      })
      .then((results) => {
        if (!results) {
          results = [];
        }
        results.forEach((func) => {
          if (func) {
            functionMap.set(
              (functions.isCFunction(func) ? func.functionName : func.mangledId),
              func
            );
          }
        });
        const functionIds = results.map(({ _id }) => _id);
        return functions.getVariablesOfFunctions(functionIds);
      })
      .then(async (functionVariables) => {
        functionVariables.forEach((functionVariable) => {
          functionVariablesMap.set(functionVariable.functionId.toString(), functionVariable);
        });
        let staticGlobalFilesMap = new Map();
        const filePathToFileIdMap = new Map();
        const funcNameToFuncIdMap = new Map();
        if (isIntegrationTest) {
          staticGlobalFilesMap = await functions.getVersionGlobalVariables(version._id, 'isStatic');
        }
        const { fclmt } = await utility.getUserDefined();
        let split = 10;
        if (fclmt > 0) {
          split = 1;
        }
        let valid = await utility.checkValidation(fclmt);
        await utility.arrayChunkOperation(content.functions, split, async ({ name, testcases }) => {
          if (!valid) {
            return;
          }
          const func = functionMap.get(name);
          const functionId = func._id;
          let functionVariables = functionVariablesMap.get(functionId.toString());
          const testsArray = [];
          const toFuncMangledIdMap = new Map();
          const toFuncNameMap = new Map();
          // 集成测试生成用例要去掉没有打桩的函数
          if (isIntegrationTest) {
            const relationFuncs = await invokeRelations.retrieveRelations(functionId);
            relationFuncs.forEach((func) => {
              if (!func.enableStub) {
                toFuncMangledIdMap.set(func.toFuncMangledId, {});
                toFuncNameMap.set(func.toFuncName, {});
              }
            });
            // 处理集成测试生成用例文件路径
            func.filePath = content.file;
            funcNameToFuncIdMap.set(functions.isCFunction(func) ? func.functionName : func.mangledId, { functionId: func._id.toString() });
            filePathToFileIdMap.set(func.filePath, { fileId: func.fileId.toString() });
            await tests.handleIntegrationAutoGenerateTestFilePath(func, testcases, filePathToFileIdMap, funcNameToFuncIdMap, staticGlobalFilesMap);
          }
          testcases.forEach((item) => {
            // c++测试用例将自动生成的手动桩函数（funcs）放到stubs中
            if (item.testcase.funcs) {
              item.testcase.stubs = item.testcase.stubs || {};
              item.testcase.stubs = { ...item.testcase.stubs };
              Object.keys(item.testcase.funcs).forEach((stubName) => {
                item.testcase.stubs[stubName] = {
                  ...item.testcase.funcs[stubName],
                  stubsTriggerKind: tests.stubTriggerKind.manual,
                };
              });
              delete item.testcase.funcs;
            }

            const { stubs } = item.testcase;
            if (isIntegrationTest) {
              if (stubs) {
                Object.keys(stubs).forEach((stubName) => {
                  // 兼容c和c++
                  if (toFuncMangledIdMap.has(stubName) || toFuncNameMap.has(stubName)) {
                    delete stubs[stubName];
                  }
                });
              }
            }
            testsArray.push(item.testcase);
          });
          if (isIntegrationTest) {
            functionVariables = await functions.getRelationFuncsVariables(func._id);
            await functions.updateFunctionVariable(functionId, testsArray, functionVariables);
            const allFileMap = await files.getSourceFileByVersionId(func.versionId, null, { _id: 1, fileName: 1, path: 1 });
            await functions.IntegrationAddInvokeFuncFixedAddressBases(
              { versionId: func.versionId, ...functionVariables },
              null,
              projectVersions.versionType.integrationTest,
              allFileMap
            );
          } else {
            await functions.updateFunctionVariable(functionId, testsArray);
          }
          await tests.replaceGeneratedTests(func, functionVariables, testcases);
          if (fclmt > 0) {
            valid = await utility.checkValidation(fclmt, true);
            const hash = await utility.getHash();
            publishValidationMessage('update', {
              creatorId: version.creatorId,
              hash,
            });
          }
        });
      })
      .then(() => {
        if (!isIntegrationTest) {
          return Promise.resolve();
        }
        const promises = [];
        functionMap.forEach((func) => {
          promises.push(functions.updateFunctionInvokeCoverage(func._id));
        });
        return Promise.all(promises);
      })
      .then(() => {
        const promises = [];
        functionMap.forEach((func) => {
          promises.push(functions.updateFunctionCoverage(func._id));
        });
        return Promise.all(promises);
      })
      .then(resolve)
      .catch(reject);
  });
}
/**
 *  处理返回结果的Callstack
 * @param {String} testId 测试用例id
 * @param {String} projectVersionId 项目版本id
 * @param {Object} content 消息内容
 * @returns
 */
function handleContentCallstack(testId, projectVersionId, content) {
  const promises = [];
  if (content.callstack) {
    promises.push(invokeRelations.invokeEdgeCovered(testId, projectVersionId, content.callstack));
  }
  // 处理编组
  if (content.resultlist && Array.isArray(content.resultlist)) {
    // eslint-disable-next-line no-inner-declarations
    function handleResultlist(resultlist) {
      if (!Array.isArray(resultlist)) {
        return;
      }
      resultlist.forEach((r) => {
        if (r.resultlist) {
          handleResultlist(r.resultlist);
        } else if (r.callstack) {
          promises.push(invokeRelations.invokeEdgeCovered(r.testcase, projectVersionId, r.callstack));
        }
      });
    }
    handleResultlist(content.resultlist);
  }
  return Promise.all(promises);
}

async function handleCoverageContent(test, taskId, projectVersion, content, coverageStatistics) {
  if (!taskCache.has(taskId.toString())) {
    taskCache.set(taskId.toString(), {});
  }
  if (!taskCache.get(taskId.toString()).lastProcessedFunc) {
    taskCache.get(taskId.toString()).lastProcessedFunc = new Set();
  }
  taskCache.get(taskId.toString()).lastProcessedFunc.add(test.functionId.toString());
  await functions.removeCfgCoverageWithTestId(null, test._id.toString());
  // storage test results
  if (coverageStatistics) {
    if (content.coverage && !utility.isEmpty(content.coverage.functions)) {
      await utility.arrayChunkOperation(content.coverage.functions, 10, async (func) => {
        let funcDb;
        if (projectVersion.versionType === projectVersions.versionType.unitTest) {
          funcDb = await functions.getFunctionByKey(test.fileId, func.mangled);
        } else {
          const file = await files.getFileByPath(projectVersion._id, func.path);
          if (file) {
            funcDb = await functions.getFunctionByKey(file._id, func.mangled);
          }
        }
        if (!funcDb) {
          return Promise.resolve();
        }
        const promises = [
          functions.updateCoveredBlocks(funcDb._id, projectVersion._id, func.blocks || []),
          functions.updateCoveredBranches(funcDb._id, projectVersion._id, func.branches || []),
          functions.updateCoveredConditions(funcDb._id, projectVersion._id, func.decisions || []),
          // 测试用例执行后补充新增的cfg数据，避免重新解析
          functions.completeCoverageInfo(funcDb._id, func),
        ];
        return Promise.all(promises);
      });
      if (projectVersion.versionType === projectVersions.versionType.integrationTest) {
        // 覆盖率计算方式变化,兼容性代码
        if (!(await functions.isCompatible(test.functionId))) {
          await functions.updateInvokeCoverageTotal([test.functionId], projectVersion._id);
        }
        await functions.updateCoveredInvokes(test._id, projectVersion._id, content.coverage);
      }
    }
  }
  return handleContentCallstack(test._id, projectVersion._id, content);
}

/**
 * 处理执行结果消息
 * @param {String} taskId 任务id
 * @param {Object} projectVersion 项目版本
 * @param {Object} content 消息内容
 * @param {Boolean} coverageStatistics 是否收集覆盖率
 * @return {Promise}
 */
async function handleResultMessage(taskId, projectVersion, content, coverageStatistics) {
  const isGroup = content.resultlist !== undefined;
  const testId = content.testcase;
  let testcaseIds = [];
  const testCase = await (isGroup ? TestGroup : Test).findById(testId, ['fileId', 'functionId']);
  const file = await File.findById(testCase.fileId, ['path']);
  const func = await Func.findById(testCase.functionId, ['mangledId']);
  // 集成测试只保留被测函数的callstack
  if (content.callstack && Object.keys(content.callstack).length > 1) {
    const key = `${func.mangledId}@${file.path}`;
    content.callstack = {
      [key]: content.callstack[key],
    };
  }
  if (testCase && file) {
    publishSingleTaskMessage('update', {
      taskId: content.phoenixTaskId,
      currentFile: path.basename(file.path),
      subTaskType: tasks.type.testCaseRun,
    });
  }
  return tests.formatTestCasesOutput(
    content,
    projectVersion,
    isGroup
  )
    .then(({ testCaseOutputMap: testCasesOutput, runtimeOutputMap, testCaseIds }) => {
      const promise = [];
      testCasesOutput.forEach((testcaseOutput, testId) => {
        promise.push(tests.updateTestOutputResult(testId, taskId, testcaseOutput, projectVersion, content.error));
      });
      runtimeOutputMap.forEach((runtime, testId) => {
        promise.push(breakpoint.updateBreakPointOutputResultByTestcase(testId, runtime));
      });
      testcaseIds = testCaseIds;
      return Promise.all(promise);
    })
    .then(() => handleCoverageContent(testCase, taskId, projectVersion, content, coverageStatistics))
    .then(() => UserDefinedFields.findOne({ domain: projectVersion.versionType, scope: 'testcase', name: '用例注释' }))
    .then((userDefinedField) => utility.arrayChunkOperation(
      testcaseIds, 10, (testcaseId) => tests.updateTestcaseComment(testcaseId, userDefinedField)
    ));
}

/**
 * 处理函数覆盖数据消息
 * @param {Object} taskInfo 任务信息
 * @param {Object} projectVersion 项目版本
 * @param {Object} content 消息内容
 * @param {Number} ackNo 任务序号
 * @return {Promise}
 */
async function handleCoverageMessage(
  taskInfo,
  projectVersion,
  content,
) {
  // publishSingleTaskMessage('update', {
  //   taskId: content.phoenixTaskId,
  //   currentFile: path.basename(content.file),
  //   subTaskType: tasks.type.testCaseRun,
  // });
  if (taskInfo.type === tasks.type.codeInstrumentationReportGen) {
    if (!utility.isArrayType(content.loggers)) {
      return Promise.resolve();
    }
    const resultId = content.loggers[0];
    await codeInstruments.storeResultCovered(content, taskInfo.codeInstrumentationId, resultId, projectVersion);
    return codeInstruments.updateResultStatus(
      taskInfo.codeInstrumentationId,
      resultId,
      codeInstruments.resultStatusKind.success
    );
  }
  const taskId = taskInfo.taskId[taskInfo.taskId.length - 1];
  if (!taskCache.has(taskId) || !taskCache.get(taskId).lastProcessedFunc) {
    return Promise.resolve();
  }
  return Promise.all([...taskCache.get(taskId).lastProcessedFunc].map(async (functionId) => {
    const func = await functions.getFunction(functionId);
    // 更新cases，算法二为全量，算法一为全量子集，算法二暂时和算法一相同处理，后期数据量太大考虑实时获取
    await functions.updateMcdcCases(func._id, projectVersion.mcdcKind);
    await functions.updateFunctionCoverage(func._id);
    if (taskInfo.type === tasks.type.autoTestAndFilter) {
      // 根据任务类型，进行测试用例的hash计算以及清除
      await functions.deleteFunctionDuplicateTestcases(func._id, projectVersion._id);
    }
    taskCache.get(taskId).lastProcessedFunc.delete(functionId.toString());
    return Promise.resolve();
  }));
}

/**
 * 处理进度消息
 * @param {String} taskId 任务id
 * @param {Object} content 消息内容
 * @param {Number} ackNo 消息序号
 * @return {Promise}
 */
async function handleProgressMessage(taskId, content, ackNo) {
  let task = await tasks.getTask(taskId);
  if (!task) {
    return Promise.resolve();
  }
  if (content.state === tasks.statusKind.finished) {
    taskCache.delete(task.taskId[task.taskId.length - 1]);
    if (task.currentSubTask === tasks.type.parse) {
      // 计算扇入删除
      await functions.replaceFanInAndFanOut(task.versionId);
    } else if (task.type === tasks.type.instrument || task.type === tasks.type.exportTargetMachineTests) {
      await codeInstruments.updateStatus(task.codeInstrumentationId, codeInstruments.statusKind.finish);
    }
    task = await tasks.updateSubTaskStatus(taskId, tasks.subStatusKind.finished);
    return tasks.runNextTask(task);
  }
  if (ackNo === 1) {
    // 静态分析任务
    if (task.type === tasks.type.analyze) {
      await defects.updateRuleStatistics(task.versionId);
    }
    if (task.status === tasks.statusKind.waiting) {
      return tasks.updateTaskStatus(task._id, content.state);
    }
  }
  return tasks.updateProgress(task._id, Number(content.progress) / 100);
}

/**
 * 处理用例执行运行时错误消息
 * @param {String} taskId 任务id
 * @param {String} testId 测试用例id
 * @param {Object} projectVersion 项目版本
 * @param {Object} content 消息内容
 * @param {Boolean} coverageStatistics 是否收集覆盖率
 * @return {Promise}
 */
async function handleRuntimeErrorMessage(taskId, testId, projectVersion, content, coverageStatistics) {
  let testIdWithError = testId;
  let isGroup = false;
  let test = await tests.getTest(testId);
  if (!test) {
    const testCaseGroup = await testCaseGroups.getGroupById(testId);
    if (testCaseGroup.length > 0) {
      isGroup = true;
    }
  }
  const { testCaseOutputMap: testCasesOutput, runtimeOutputMap } = await tests.formatTestCasesOutput(content, projectVersion, isGroup);
  let promise = [];
  if (isGroup) {
    testIdWithError = null;
    const testCaseIds = await testCaseGroups.getAllTestsByGroupId(testId);
    testCaseIds.forEach((testCaseId) => {
      if (testIdWithError) {
        return;
      }
      if (!testCasesOutput.has(testCaseId)) {
        testIdWithError = testCaseId;
      }
    });
    testCasesOutput.forEach((testcaseOutput, testId) => {
      promise.push(tests.updateTestOutputResult(
        testId,
        taskId,
        testcaseOutput,
        projectVersion
      ));
    });
    await Promise.all(promise);
  }
  promise = [];
  runtimeOutputMap.forEach((runtime, testId) => {
    promise.push(breakpoint.updateBreakPointOutputResultByTestcase(testId, runtime));
  });
  await Promise.all(promise);
  if (!testIdWithError) {
    return Promise.resolve();
  }
  let resultKind;
  switch (content.header) {
    case 'Runtime Error':
      resultKind = tests.resultKind.runtimeError;
      break;
    case 'Time Limit Exceeded':
      resultKind = tests.resultKind.timeout;
      break;
    case 'System Error':
      resultKind = tests.resultKind.systemError;
      break;
    case 'Instrument Error':
      resultKind = tests.resultKind.instrumentError;
      break;
    case 'Compile Error':
      resultKind = tests.resultKind.compileError;
      break;
    default:
      return Promise.resolve();
  }
  if (testIdWithError.toString() !== testId.toString()) {
    test = await tests.getTest(testIdWithError);
  }
  await tests.handleRuntimeErrorResult(
    testIdWithError,
    taskId,
    resultKind,
    content.content,
    content.location,
    projectVersion,
  );
  return handleCoverageContent(test, taskId, projectVersion, content, coverageStatistics);
}

async function handleTestcaseMigrationMessage(taskInfo, projectVersion, content) {
  const { functionId } = content;
  if (taskInfo.status === tasks.statusKind.canceled) {
    return;
  }
  const taskId = taskInfo.taskId[taskInfo.taskId.length - 1];
  if (!projectVersion.oldVersionId) {
    return;
  }
  const func = await functions.getFunction(functionId);
  if (!func) {
    return;
  }
  if (func.taskId !== taskId) {
    return;
  }
  // clear testcases
  const testCases = await tests.getTests({}, { functionId });
  await utility.promiseChain(
    testCases.map((testCase) => () => tests.removeTest(testCase._id, testCase))
  );
  await tests.reshuffleTestNumber(functionId);
  await testCaseGroups.reshuffleGroupNumber(functionId);
  await functions.updateFunctionCoverage(functionId);
  let resVersionId = projectVersion.oldVersionId;
  if (projectVersion.versionType === projectVersions.versionType.integrationTest) {
    await functions.updateFunctionInvokeCoverage(functionId);
    const resVersion = await projectVersions.initSubVersion(projectVersion.oldVersionId,
      projectVersions.versionType.integrationTest);
    resVersionId = resVersion._id;
  }
  await projectVersions.migrateTestCaseOfVersion(
    resVersionId,
    {
      functionIds: [functionId],
      targetVersionId: projectVersion._id,
      creator: projectVersion.creatorId,
      isRegressionReusedCases: true,
    }
  );
}

async function handleCouplingMessage(projectVersion, content) {
  const file = await files.getFileByPath(projectVersion._id, content.file);
  if (!file) {
    return;
  }
  await DataFlow.deleteMany({ fileId: file._id });
  if (!content.Result || content.Result.length === 0) {
    return;
  }
  await Promise.all(utility.arrayChunk(content.Result, 500).map(
    (couplings) => {
      couplings.forEach((coupling) => {
        coupling.creatorId = projectVersion.creatorId;
        coupling.projectId = projectVersion.projectId;
        coupling.versionId = projectVersion._id;
        coupling.fileId = file._id;
      });
      return DataFlow.collection.insertMany(couplings, { ordered: false }).catch(() => { });
    }
  ));
}
/**
 * 处理testbed信息流数据信息
 * @param {Object} projectVersion 版本
 * @param {Object} content 消息内容
 */
async function handleTestbedFlowMessage(projectVersion, content) {
  const file = await files.getFileByPath(projectVersion._id, content.file);
  if (!file || !content.rule) {
    return;
  }
  const fileId = file._id.toString();
  const filePath = file.path;
  const ruleType = content.rule;
  await testbeds.deleteFlow(file._id, { ruleType });
  const funcs = await Func.find({ fileId }, ['_id', 'mangledId']).lean();
  if (funcs.length === 0) {
    return;
  }
  const funcMangledIdMap = new Map();
  const filePathMap = new Map();
  filePathMap.set(filePath, fileId);
  funcs.forEach((func) => {
    funcMangledIdMap.set(func.mangledId, func._id.toString());
  });
  switch (ruleType) {
    case 'PathTracker':
    {
      await testbeds.handlePathTrackerMessage(content.functions, {
        fileId,
        filePathMap,
        funcMangledIdMap,
        projectVersion,
      });
      break;
    }
    case 'DependencyAnalyzer': {
      await testbeds.handleDependencyAnalyzerMessage(content.functions, {
        fileId,
        funcMangledIdMap,
        projectVersion,
      });
      break;
    }
    case 'Statistics': {
      await testbeds.handleStatisticsMessage(content.functions, {
        fileId,
        funcMangledIdMap,
        projectVersion,
      });
      break;
    }
    case 'DateObjectAnalyzer': {
      await testbeds.handleDateObjectAnalyzerMessage(content.objects, {
        fileId,
        projectVersion,
      });
      break;
    }
    case 'LifetimeAnalyzer': {
      await testbeds.handleLifetimeAnalyzerMessage(content.vars, {
        fileId,
        filePathMap,
        funcMangledIdMap,
        projectVersion,
      });
      break;
    }
    default: break;
  }
}

async function handleInstrumentMessage(taskInfo, filePath, projectVersion, content) {
  const { codeInstrumentationId } = taskInfo;
  const dir = path.join(cacheUtil.getCodeInstrumentationDir(), projectVersion._id.toString(), codeInstrumentationId.toString());
  const fullPath = path.join(dir, filePath);
  await fs.ensureDir(path.dirname(fullPath));
  await fs.writeFile(fullPath, content);
}

/**
 * 处理消息
 * @param {Object} msg 消息内容
 */
function handleMessage(msg) {
  const taskId = msg.properties.headers['task-id'];
  const messageType = msg.properties.headers['message-type'];
  const projectId = msg.properties.headers['project-id'];
  const testId = msg.properties.headers.testcase;
  const ackNo = Number(msg.properties.headers['ack-no']);
  const filePath = msg.properties.headers.file;
  // 临时解决方案，后续需新增字段
  const phoenixTaskId = msg.properties.headers['tester-id'];
  logger.debug(`messageType:${messageType} TaskId:${taskId} ackNo:${ackNo} phoenixTaskId:${phoenixTaskId}`);
  let content = msg.content.toString();
  const contentStr = content;
  let taskInfo;
  let taskLog;
  let projectVersionId;
  return Promise.resolve()
    .then(async () => {
      if (!taskId || Number.isNaN(ackNo)) {
        throw new Error('need taskId and ackNo');
      }
      taskLog = await TaskLog.findOneAndUpdate({
        taskId,
        ackNo,
      }, {
        $set: {
          projectId,
          taskId,
          phoenixTaskId,
          type: messageType,
          status: 'pending',
          ackNo,
          modifiedAt: new Date(),
        },
        $setOnInsert: {
          content: content.substring(0, 65536),
          createdAt: new Date(),
        },
        $inc: { tryCnt: 1 },
      }, {
        upsert: true,
        new: true,
      }).select({ content: 0 });
      if (taskLog.tryCnt > MAX_TRY_CNT) {
        // 由于消息处理过程中进程崩溃导致无法ack；正常的能捕获到的错误会直接ack，不会走到这里
        throw new Error('exceeded maximum retry count');
      }
    })
    .then(() => {
      if (ackNo === 1) {
        return new Promise((resolve) => setTimeout(resolve, 1000));
      }
      return Promise.resolve();
    })
    .then(() => {
      if (messageType !== 'instrument') {
        try {
          if (messageType !== 'export') {
            content = JSON.parse(content);
          }
        } catch (e) {
          logger.verbose(content);
          logger.error('Fail to parse message content to JSON');
          throw e;
        }
      }
      return tasks.getTask(phoenixTaskId)
        .then((task) => {
          if (!task) {
            return Promise.resolve();
          }
          // 任务取消之后忽略掉该任务后续的消息
          if (task && task.status === tasks.statusKind.canceled) {
            throw new Error('task canceled');
          }
          taskInfo = task;
          content.phoenixTaskId = task._id;
          return projectVersions.getVersion(task.versionId);
        })
        .then((projectVersion) => {
          if (!projectVersion) {
            return Promise.resolve();
          }
          projectVersionId = projectVersion._id.toString();
          switch (messageType) {
            case 'testif':
              return handleTestIFMessage(taskId, projectVersionId, content);
            case 'cfg':
              return handleCfgMessage(taskId, projectVersion, content);
            case 'diag':
              return handleDiagMessage(taskId, projectVersionId, content);
            case 'prove':
              return handleProveMessage(taskId, projectVersionId, content, taskInfo);
            case 'testcase':
              return handleTestCaseMessage(projectVersion, content);
            case 'result':
              return handleResultMessage(taskId, projectVersion, content, taskInfo.coverageStatistics);
            case 'coverage':
              return handleCoverageMessage(
                taskInfo,
                projectVersion,
                content
              );
            case 'progress':
              // if (content.state === tasks.statusKind.finished) {
              //   return tasks.updateSubTaskStatus(phoenixTaskId, tasks.subStatusKind.finished).then((task) => tasks.runNextTask(task));
              // }
              return handleProgressMessage(phoenixTaskId, content, ackNo);
            case 'wukong':
              return handleAnalyzeMessage(taskId, projectVersionId, content);
            case 'hcscli':
              return handleHcscliAnalyzeMessage(taskId, projectVersionId, content);
            case 'scanner':
              return handleScannerAnalyzeMessage(taskId, projectVersionId, content);
            case 'certify':
              return handleCertifyAnalyzeMessage(taskId, projectVersionId, content);
            case 'verify':
              return handleVerifyAnalyzeMessage(taskId, projectVersionId, content);
            case 'javacheck':
              return handleJavacheckAnalyzeMessage(taskId, projectVersionId, content);
            case 'cppcheck':
              return handleCppCheckAnalyzeMessage(taskId, projectVersionId, content);
            case 'export':
              return handleTargetMachineExport(taskInfo, projectId, projectVersionId, testId, ackNo, content);
            case 'migrate-testcase':
              return handleTestcaseMigrationMessage(taskInfo, projectVersion, content);
            case 'testbed_flow':
              return handleTestbedFlowMessage(projectVersion, content);
            case 'coupling':
              return handleCouplingMessage(projectVersion, content);
            case 'instrument':
              return handleInstrumentMessage(taskInfo, filePath, projectVersion, contentStr);
            case 'typesys':
              return handleFileTypesystem(taskId, projectVersionId, content);
            case 'error':
            default:
              // errors in target machine exporting should not change the status of testcases.
              if (testId && taskInfo.type !== tasks.type.exportTargetMachineTests) {
                return handleRuntimeErrorMessage(taskId, testId, projectVersion, content, taskInfo.coverageStatistics);
              }
              logger.error(`Fail to handle ${messageType} message\nmsg:${JSON.stringify(msg)}}`);
              return Promise.resolve();
          }
        });
    })
    .then(async () => {
      if (taskLog) {
        // 消息处理成功
        taskLog.status = 'succeed';
        taskLog.versionId = projectVersionId;
        taskLog.modifiedAt = new Date();
        await taskLog.save();
      }
    }, async (err) => {
      if (taskLog) {
        // 消息处理失败
        taskLog.status = err.message === 'task canceled' ? 'canceled' : 'failed';
        taskLog.versionId = projectVersionId;
        taskLog.errMsg = err.stack || err.toString();
        taskLog.modifiedAt = new Date();
        await taskLog.save();
      }
      if (err.message === 'task canceled') {
        return Promise.resolve();
      }
      throw err;
    });
}

// /**
//  * 接收消息
//  */
// function listening() {
//   function retry() {
//     debug('The connection will be tried again in 10 seconds...');
//     setTimeout(() => listening(), 10000);
//   }
//   const { queue } = global.config.amqp;
//   MessageBus.instance = null;
//   const messagebus = MessageBus.getInstance();
//   messagebus.listen(queue, (msg) => {
//     if (!msg) {
//       return Promise.resolve();
//     }
//     return new Promise((resolve) => {
//       cls.createRequestContext({ id: uuidv4() });
//       return handleMessage(msg)
//         .catch((err) => {
//           const messageType = msg.properties.headers['message-type'];
//           const taskId = msg.properties.headers['task-id'];
//           logger.error(`Fail to handle message\ntaskId: ${taskId}\nmessageType: ${messageType}\nerror: ${err.stack}`);
//         })
//         .then(resolve);
//     });
//   })
//     .catch(() => {
//       retry();
//     });
//   messagebus.on('connection error', () => {
//     retry();
//   });
// }

function getTaskConsumerId(task) {
  const taskId = task._id.toString();
  const consumerCnt = clusterConfig[workerKind.consumer];
  // todo: 简单的一致性hash算法，后续改进
  const consumerId = parseInt(utility.md5(taskId).slice(-2), 16) % consumerCnt;
  return consumerId;
}

function getTaskQueue(taskId) {
  return `rocket-queue-${taskId}`;
}

async function rescCheck(messagebus) {
  const pendingTasks = await tasks.getPendingTasks();
  const queues = pendingTasks.filter((task) => {
    const consumerId = getTaskConsumerId(task);
    return consumerId === process.workerId;
  }).map((task) => getTaskQueue(task._id));
  if (process.workerId === 0) {
    queues.push(global.config.amqp.queue);
  }
  try {
    // eslint-disable-next-line no-restricted-syntax
    for (const queue of queues) {
      if (!messagebus.queueMap.has(queue)) {
        // eslint-disable-next-line no-await-in-loop
        await messagebus.listenTaskQueue(queue);
      }
    }
    const currentQueueMap = new Set(queues);
    // eslint-disable-next-line no-restricted-syntax
    for (const queue of [...messagebus.queueMap.keys()]) {
      if (!currentQueueMap.has(queue)) {
        // eslint-disable-next-line no-await-in-loop
        await messagebus.unlisten(queue);
      }
    }
  } catch (err) {
    logger.error(err);
    process.exit(1);
  }
  setTimeout(() => rescCheck(messagebus), 5000);
}

/**
 * 1. 读取数据库中所有进行中的task，并选取自身负责的task进行队列消费
 * 2. 监听EVENT_NEW_TASK事件，如果新的task归自身负责，则对其相应的队列进行消费
 * 3. 监听EVENT_TASK_QUEUE_REMOVED事件，并对相应的队列移出监听
 * 注意!!!: 此方法仅进程启动时调用一次，不可重复调用
 */
async function listening() {
  MessageBus.instance = null;
  const messagebus = MessageBus.getInstance();
  messagebus.on('connection error', () => {
    process.exit(1);
  });
  setTimeout(() => rescCheck(messagebus), 1);
}

function sendToQueue(queue, message, options = {}) {
  return MessageBus.getInstance().sendToQueue(queue, message, options);
}

async function createTaskQueue(taskId) {
  return MessageBus.getInstance().createTaskQueue(taskId);
}

async function removeQueue(queue) {
  return MessageBus.getInstance().removeQueue(queue);
}
/**
 * Export model definition object.
 */
module.exports = {
  checkConnection,
  listening,
  sendToQueue,
  MessageBus,
  createTaskQueue,
  removeQueue,
  getTaskQueue,
};
exports = Object.assign(exports, module.exports);
